package com.electric.flinkcep.serialization;

import com.electric.flinkcep.model.PredictData;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * PredictData 反序列化器
 * 用于从Kafka消息中反序列化预测电力负荷数据
 */
public class PredictDataDeserializer implements DeserializationSchema<PredictData> {
    
    private static final Logger logger = LoggerFactory.getLogger(PredictDataDeserializer.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public PredictData deserialize(byte[] message) throws IOException {
        try {
            String jsonString = new String(message, StandardCharsets.UTF_8);
            logger.debug("反序列化PredictData消息: {}", jsonString);
            
            PredictData predictData = objectMapper.readValue(jsonString, PredictData.class);
            logger.debug("成功反序列化PredictData: {}", predictData);
            
            return predictData;
        } catch (Exception e) {
            logger.warn("反序列化PredictData失败，跳过消息: {}", new String(message, StandardCharsets.UTF_8));
            // 抛出异常让Flink跳过这条消息
            throw new IOException("Invalid PredictData format", e);
        }
    }
    
    @Override
    public boolean isEndOfStream(PredictData nextElement) {
        return false;
    }
    
    @Override
    public TypeInformation<PredictData> getProducedType() {
        return TypeInformation.of(PredictData.class);
    }
}